package com.vincent;

import com.sun.istack.internal.NotNull;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;

import javax.jms.*;
import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by vincent on 2018/7/29.
 * activeMQ消费者测试,pull方式
 */
public class ConsumerTest {
    private static Map<String, MessageConsumer> consumerMap = new ConcurrentHashMap<>();

    public static void main(String[] args) throws JMSException, InterruptedException {
//        String url = "tcp://192.168.1.2:61616";
//        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
//        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);
//        for (int i = 0; i < 3; i++) {
//            Thread thread = new Thread(new ConsumerRunner(pooledConnectionFactory, "test-queue", i + ""));
//            thread.setName("thread-"+i);
//            thread.start();
//        }

/*        while (true) {

            String s = receiveMQ("test-queue");
            System.out.println("收到消息:" + s);
            Thread.sleep(1000L);
        }*/
        ExecutorService executorService = new ThreadPoolExecutor(3, 8, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        for (int i = 0; i < 3; i++) {
            System.out.println("----------------");
            executorService.execute(new ConsumerReceiveRunner());
        }

    }

    public static String receiveMQ(@NotNull String topicName) throws JMSException {
        //想要平均消费必须配置jms.prefetchPolicy，默认值queue是1000,topic是100，此处修改为0
        if (!consumerMap.containsKey(topicName)) {
            //String url = "tcp://192.168.1.2:61616?jms.prefetchPolicy.queuePrefetch=0";
            String url = "tcp://192.168.1.2:61616";
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);
            PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(connectionFactory);
            Connection connection = pooledConnectionFactory.createConnection();
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = new ActiveMQQueue("test-queue?consumer.prefetchSize=0");
            MessageConsumer consumer = session.createConsumer(destination);
            consumerMap.put(topicName, consumer);
        }
        MessageConsumer messageConsumer = consumerMap.get(topicName);
        TextMessage message = (TextMessage) messageConsumer.receive();
        if (message != null)
            return message.getText();
        else
            return null;
    }
}
